package com.flitsneak.redis.config;

import com.flitsneak.common.entity.Info;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.ReactiveRedisTemplate;
import org.springframework.data.redis.core.ReactiveStreamOperations;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.Subscription;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.concurrent.Executors;

/**
 * @author flitsneak
 * @email flitsneak@gmail.com
 * @date 2022/8/10 15:55
 * @description
 */
@Slf4j
@RequiredArgsConstructor
@Configuration
public class RedisStreamConfig {

    private static final String STREAM_GROUP = "good-group";

    private static final String STREAM_KEY = "good-stream";

    private static final String CONSUMER_A = "consumer-a";

    private final StreamListener<String, ObjectRecord<String, Info>> streamListener;

    private final ReactiveRedisTemplate<String, String> reactiveRedisTemplate;

    @Bean
    public Subscription subscription(RedisConnectionFactory redisConnectionFactory) {
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ObjectRecord<String, Info>> options = StreamMessageListenerContainer
                .StreamMessageListenerContainerOptions
                .builder()
                .batchSize(100)
                .executor(Executors.newSingleThreadExecutor())
                .pollTimeout(Duration.ofSeconds(2)) // 超时时间，如果为0表示不设置超时时间，如果为正数，一旦超时则会抛出异常
                .targetType(Info.class)
                .errorHandler(e-> log.error("redis stream 异常：{}",e.getMessage()))
                .build();
        StreamMessageListenerContainer<String, ObjectRecord<String, Info>> listenerContainer = StreamMessageListenerContainer.create(redisConnectionFactory, options);

        return prepareStreamAndGroup(reactiveRedisTemplate.opsForStream(), STREAM_KEY, STREAM_GROUP)
                .map(s -> {
                    //自动签收 listenerContainer.receiveAutoAck
                    //手动签收
                    Subscription subscription = listenerContainer.receive(
                            Consumer.from(STREAM_GROUP, CONSUMER_A),
                            StreamOffset.create(STREAM_KEY, ReadOffset.lastConsumed()),
                            streamListener);
                    listenerContainer.start();
                    return subscription;
                }).block();
    }

    /**
     * 查找Stream信息，如果不存在，则创建Stream
     */
    private Mono<StreamInfo.XInfoStream> prepareStreamAndGroup(ReactiveStreamOperations<String, ?, ?> ops, String stream, String group) {
        // info方法查询Stream信息，如果该Stream不存在，底层会报错，这时会调用onErrorResume方法。
        return ops.info(stream)
                .onErrorResume(err -> {
                    log.warn("获取stream信息异常:{}", err.getMessage());
                    //创建Stream
                    return ops.createGroup(stream, group)
                            .flatMap(s -> ops.info(stream));
                });
    }


}
